package com.yujiahao.bigdata.rdd.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
    todo 案例实操
      1)	数据准备
      agent.log：时间戳，省份，城市，用户，广告，中间字段使用空格分隔。
      2)	需求描述
      统计出每一个省份每个广告被点击数量排行的Top3
      3)	需求分析
      4)	功能实现

 */

object Spark_RDD1_To_Rdd2_Pper_Transform_GroupByKeyTest {
  def main(args: Array[String]): Unit = {


    //TODO 1、获得连接
    val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 3、需求：统计出每一个省份每个广告被点击数量排行的Top3
    //先聚合，在分组（word Count）
    //todo 1)读取数据文件
    val rdd: RDD[String] = sc.textFile("data/agent.log")
    //todo 2)将读取的数据进行格式转换，用于统计分析
    val wordCountRDD: RDD[((String, String), Int)] = rdd.map(

      line => {
        val datas: Array[String] = line.split(" ")
        ((datas(1), datas(4)), 1)
      }
    ).reduceByKey(_ + _)

    //todo 3)将统计后的结果进行结构转换

    val mapRDD: RDD[(String, (String, Int))] = wordCountRDD.map {
      case ((prv, ad), sum) => {
        (prv, (ad, sum))
      }
    }
    //todo 4)将转换后的结果进行分组
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = mapRDD.groupByKey()
    //todo 5)将分组后的数据排序
    val top3: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
      }
    )
    //todo 6)打印
    top3.collect().foreach(println)

    //TODO 2、关闭连接
    sc.stop()
  }
}